package cn.chengpengper.rpc.registry.nacos;

import cn.chengpengper.rpc.common.helper.RpcServiceHelper;
import cn.chengpengper.rpc.constants.RpcConstants;
import cn.chengpengper.rpc.loadbalancer.api.ServiceLoadBalancer;
import cn.chengpengper.rpc.loadbalancer.helper.ServiceLoadBalancerHelper;
import cn.chengpengper.rpc.protocol.meta.ServiceMeta;
import cn.chengpengper.rpc.registry.api.RegistryService;
import cn.chengpengper.rpc.registry.api.config.RegistryConfig;
import cn.chengpengper.rpc.spi.annotation.SpiClass;
import cn.chengpengper.rpc.spi.loader.ExtensionLoader;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.lang.TypeReference;
import cn.hutool.json.JSONUtil;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.curator.x.discovery.ServiceInstance;

import java.io.IOException;
import java.util.*;

/**
 * @author chengpeng.hu
 */
@Slf4j
@SpiClass
public class NacosRegistryServiceImpl implements RegistryService {

    private NamingService naming;

    private ServiceLoadBalancer<ServiceMeta> serviceLoadBalancer;

    private ServiceLoadBalancer<ServiceMeta> serviceEnhancedLoadBalancer;

    @Override
    public void init(RegistryConfig registryConfig) throws Exception {
        log.info("基于Nacos的注册中心");
        naming = NamingFactory.createNamingService(registryConfig.getRegistryAddr());

        //增强型负载均衡策略
        if (registryConfig.getRegistryLoadBalanceType().toLowerCase().contains(RpcConstants.SERVICE_ENHANCED_LOAD_BALANCER_PREFIX)) {
            this.serviceEnhancedLoadBalancer = ExtensionLoader.getExtension(ServiceLoadBalancer.class, registryConfig.getRegistryLoadBalanceType());
        } else {
            this.serviceLoadBalancer = ExtensionLoader.getExtension(ServiceLoadBalancer.class, registryConfig.getRegistryLoadBalanceType());
        }
    }

    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        Instance instance = new Instance();
        instance.setIp(serviceMeta.getServiceAddr());
        instance.setPort(serviceMeta.getServicePort());
        instance.setHealthy(true);
        instance.setWeight(serviceMeta.getWeight());
        instance.setServiceName(serviceMeta.getServiceName());

        Map<String, String> meta = new HashMap<>(8);
        meta.put("serviceName", serviceMeta.getServiceName());
        meta.put("serviceVersion", serviceMeta.getServiceVersion());
        meta.put("serviceAddr", serviceMeta.getServiceAddr());
        meta.put("servicePort", String.valueOf(serviceMeta.getServicePort()));
        meta.put("weight", String.valueOf(serviceMeta.getWeight()));
        meta.put("serviceGroup", serviceMeta.getServiceGroup());
        instance.setMetadata(meta);

        naming.registerInstance(RpcServiceHelper.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion(), serviceMeta.getServiceGroup()), instance);
    }

    @Override
    public void unRegister(ServiceMeta serviceMeta) throws Exception {
        naming.deregisterInstance(RpcServiceHelper.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion(), serviceMeta.getServiceGroup()), serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
    }

    @Override
    public ServiceMeta select(List<ServiceMeta> serviceMetaList, int invokerHashCode, String sourceIp) {
        return this.serviceLoadBalancer.select(serviceMetaList, invokerHashCode, sourceIp);
    }

    @Override
    public ServiceMeta discovery(String serviceName, int invokerHashCode, String sourceIp) throws Exception {
        // 获取健康的实例列表
        List<Instance> healthyInstances = naming.selectInstances(serviceName, true);

        if (serviceLoadBalancer != null) {
            return getServiceMetaInstance(invokerHashCode, sourceIp, healthyInstances);
        }

        List<ServiceInstance<ServiceMeta>> serviceInstances = new ArrayList<>();
        for (Instance instance : healthyInstances) {
            ServiceInstance<ServiceMeta> item = ServiceInstance.<ServiceMeta>builder()
                    .address(instance.getIp())
                    .port(instance.getPort())
                    .name(instance.getServiceName())
                    .id(instance.getInstanceId())
                    .payload(convert(instance))
                    .build();
            serviceInstances.add(item);
        }
        return this.serviceEnhancedLoadBalancer.select(ServiceLoadBalancerHelper.getServiceMetaList(serviceInstances), invokerHashCode, sourceIp);
    }

    @Override
    public List<ServiceMeta> discoveryAll() throws Exception {
        List<ServiceMeta> list = new ArrayList<>();
        // 这里不太对，Nacos无法获取所有服务实例，只能根据serviceName查找
        List<ServiceInfo> subscribeServices = naming.getSubscribeServices();
        if(CollectionUtils.isEmpty(subscribeServices)){
            return list;
        }
        for (ServiceInfo service : subscribeServices) {
            List<Instance> instances = naming.selectInstances(service.getName(), true);
            for (Instance instance : instances) {
                list.add(convert(instance));
            }
        }
        return list;
    }

    private ServiceMeta getServiceMetaInstance(int invokerHashCode, String sourceIp, List<Instance> serviceInstances) {
        ServiceMeta instance = this.serviceLoadBalancer.select(convert(serviceInstances), invokerHashCode, sourceIp);
        if (instance != null) {
            return instance;
        }
        return null;
    }

    private List<ServiceMeta> convert(List<Instance> serviceInstances) {
        List<ServiceMeta> serviceMetaList = new ArrayList<>();
        for (Instance instance : serviceInstances) {
            serviceMetaList.add(convert(instance));
        }
        return serviceMetaList;
    }

    private ServiceMeta convert(Instance instance) {
        Map<String, String> metadata = instance.getMetadata();

        ServiceMeta meta = new ServiceMeta();
        meta.setServiceName(metadata.get("serviceName"));
        meta.setServiceAddr(metadata.get("serviceAddr"));
        meta.setServicePort(Integer.parseInt(metadata.get("servicePort")));
        meta.setServiceGroup(metadata.get("serviceGroup"));
        meta.setServiceVersion(metadata.get("serviceVersion"));
        meta.setWeight(((int) instance.getWeight()));
        return meta;
    }

    @Override
    public void destroy() throws IOException {
        try {
            naming.shutDown();
        } catch (NacosException e) {
            log.error("destroy has a exception");
        }
    }
}
